package com.fwmagic.flink.parametertool;

import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.shaded.org.joda.time.DateTime;

import java.util.HashMap;
import java.util.Map;

public class ParameterToolTest {
    public static void main(String[] args) throws Exception{
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        ParameterTool parameterTool1 = ParameterTool.fromPropertiesFile(new File("/application.properties"));
        ParameterTool parameterTool2 = ParameterTool.fromPropertiesFile("/application.properties");
        ParameterTool parameterTool3 = ParameterTool.fromSystemProperties()
                .mergeWith(ParameterTool.fromPropertiesFile(ParameterToolTest.class.getResourceAsStream("/application.properties")));
*/
        DataStreamSource<String> dataSource = env.fromElements("a", "b", "c");
        dataSource.print();


        ParameterTool parameterTool = ParameterTool.fromArgs(args)
                .mergeWith(ParameterTool.fromPropertiesFile(ParameterToolTest.class.getResourceAsStream("/conf.properties")));
        System.out.println(parameterTool.get("name"));
        System.out.println(parameterTool.get("age"));
        /**
         * 注册全局参数
         * ParameterTool parameterTool = (ParameterTool) getRuntimeContext().getExecutionConfig().getGlobalJobParameters();
         *  context.collect(System.currentTimeMillis() + parameterTool.get("os.name") + parameterTool.get("user.home"));
         */
        env.getConfig().setGlobalJobParameters(parameterTool);

        env.execute();
    }

    public static ParameterTool createParameterTool(final String[] args) throws Exception {
        return ParameterTool
                .fromPropertiesFile(ParameterToolTest.class.getResourceAsStream("/application.properties"))
                .mergeWith(ParameterTool.fromArgs(args))
                .mergeWith(ParameterTool.fromSystemProperties())
                .mergeWith(ParameterTool.fromMap(getenv()));// mergeWith 会使用最新的配置
    }

    //获取 Job 设置的环境变量
    private static Map<String, String> getenv() {
        Map<String, String> map = new HashMap<>();
        for (Map.Entry<String, String> entry : System.getenv().entrySet()) {
            map.put(entry.getKey().toLowerCase().replace('_', '.'), entry.getValue());
        }
        return map;
    }
}
